package client.kafka.impl;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Properties;
import java.util.concurrent.ExecutionException;

import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.ConsumerGroupDescription;
import org.apache.kafka.clients.admin.ConsumerGroupListing;
import org.apache.kafka.clients.admin.DescribeClusterResult;
import org.apache.kafka.clients.admin.KafkaAdminClient;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.clients.admin.TopicDescription;
import org.apache.kafka.clients.admin.TopicListing;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import client.kafka.IKafkaClient;
import core.AlertUtil;
import core.FormDialog;
import javafx.scene.control.TextArea;
import kafka.admin.AclCommand;
import kafka.admin.BrokerApiVersionsCommand;
import kafka.admin.ConfigCommand;
import kafka.admin.ConsumerGroupCommand;
import kafka.admin.DelegationTokenCommand;
import kafka.admin.DeleteRecordsCommand;
import kafka.admin.LogDirsCommand;
import kafka.admin.PreferredReplicaLeaderElectionCommand;
import kafka.admin.ReassignPartitionsCommand;
import kafka.admin.TopicCommand;
import kafka.tools.ConsoleConsumer;
import kafka.tools.ConsoleProducer;
import kafka.tools.ConsumerPerformance.ConsumerPerfConfig;
import kafka.tools.DumpLogSegments;
import kafka.tools.MirrorMaker;
import kafka.tools.ReplicaVerificationTool;
import kafka.tools.StreamsResetter;
import util.JavaProcessBuilder;

/**
 * 单台Kafka连接客户端
 * @author self
 *
 */
public class KafkaClientImpl implements IKafkaClient {

	private static final Logger log=LoggerFactory.getLogger(KafkaClientImpl.class);

	/**
	 * 命令行支持的命令
	 */
	private static final Map<String,Class<?>> KAFKA_COMMAND = new HashMap<String,Class<?>>();
	static {
		KAFKA_COMMAND.put("acls", AclCommand.class);
		KAFKA_COMMAND.put("broker-api-versions", BrokerApiVersionsCommand.class);
		KAFKA_COMMAND.put("configs", ConfigCommand.class);
		KAFKA_COMMAND.put("console-consumer", ConsoleConsumer.class);
		KAFKA_COMMAND.put("console-producer", ConsoleProducer.class);
		KAFKA_COMMAND.put("consumer-groups", ConsumerGroupCommand.class);
		KAFKA_COMMAND.put("consumer-perf-test", ConsumerPerfConfig.class);
		KAFKA_COMMAND.put("delegation-tokens", DelegationTokenCommand.class);
		KAFKA_COMMAND.put("delete-records", DeleteRecordsCommand.class);
		KAFKA_COMMAND.put("dump-log", DumpLogSegments.class);
		KAFKA_COMMAND.put("log-dirs", LogDirsCommand.class);
		KAFKA_COMMAND.put("mirror-maker", MirrorMaker.class);
		KAFKA_COMMAND.put("preferred-replica-election", PreferredReplicaLeaderElectionCommand.class);
		KAFKA_COMMAND.put("reassign-partitions", ReassignPartitionsCommand.class);
		KAFKA_COMMAND.put("replica-verification", ReplicaVerificationTool.class);
		KAFKA_COMMAND.put("streams-application-reset", StreamsResetter.class);
		KAFKA_COMMAND.put("topics", TopicCommand.class);
	}

	//客户端名称
	private static final String CLIENT_NAME = "Kafka客户端";

	//管理客户端
	private AdminClient adminClient;

	//连接信息
	private String connStr;

	@Override
	public String getName() {
		return CLIENT_NAME;
	}

	@Override
	public String getConnStr() {
		return this.connStr;
	}

	@Override
	public boolean init() {
		log.info("初始化客户端组件！");
		return createConnection();
	}

	/**
	 * 创建连接
	 * @return
	 */
	private boolean createConnection() {
		log.info("创建Socket连接！");
		TextArea serverTextArea = new TextArea("127.0.0.1:9092");
		FormDialog newConnForm = new FormDialog("新建链接"
				,new FormDialog.Item("主机：", serverTextArea));
		Optional<Boolean> result = newConnForm.showAndWait();
		if(result.isPresent()) {
			String servers = serverTextArea.getText();
			log.info("Socket连接参数 -> "+servers);
			this.connStr = servers;
			Properties kafkaProp = new Properties();
			kafkaProp.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, connStr);
			adminClient = KafkaAdminClient.create(kafkaProp);
			log.info("已连接到Kafka服务器。");
			return true;
		}
		return false;
	}

	@Override
	public void createTopics(List<NewTopic> topicList) {
		adminClient.createTopics(topicList);
	}

	@Override
	public void deleteTopics(List<String> topicList) {
		adminClient.deleteTopics(topicList);
	}

	@Override
	public Collection<TopicListing> listTopics() {
		try {
			return adminClient.listTopics().listings().get();
		} catch (InterruptedException | ExecutionException e) {
			throw new RuntimeException(e);
		}
	}

	@Override
	public Collection<ConsumerGroupListing> listConsumerGroups() {
		try {
			return adminClient.listConsumerGroups().all().get();
		} catch (InterruptedException | ExecutionException e) {
			throw new RuntimeException(e);
		}
	}

	@Override
	public DescribeClusterResult describeCluster() {
		return adminClient.describeCluster();
	}

	@Override
	public TopicDescription describeTopics(String topic) {
		try {
			return adminClient.describeTopics(Arrays.asList(topic)).values().get(topic).get();
		} catch (InterruptedException | ExecutionException e) {
			throw new RuntimeException(e);
		}
	}

	@Override
	public ConsumerGroupDescription describeConsumerGroups(String groupId) {
		try {
			return adminClient.describeConsumerGroups(Arrays.asList(groupId)).describedGroups().get(groupId).get();
		} catch (InterruptedException | ExecutionException e) {
			throw new RuntimeException(e);
		}
	}

	@Override
	public String exeCmd(String cmd) {
		if(cmd==null||"".equals(cmd.trim())) {
			return null;
		}
		log.info("执行命令："+cmd);
		//命令如果没有换行符则补充一个
		if(!cmd.endsWith("\n")) {
			cmd+="\n";
		}

		//如果没有设定连接的kafka服务，添加当前连接
		if(!cmd.contains("--bootstrap-server") && !cmd.contains("--zookeeper")) {
			cmd+=" --bootstrap-server "+this.connStr;
		}

		//分割出命令行参数
		String[] cmdAndArgs = cmd.trim().split("\\s+");
		String command = cmdAndArgs[0];
		String[] args = Arrays.copyOfRange(cmdAndArgs, 1, cmdAndArgs.length);
		StringBuilder resultBuilder = new StringBuilder();
		if(KAFKA_COMMAND.containsKey(command)) {
			try {
				ProcessBuilder processBuilder = JavaProcessBuilder.build(KAFKA_COMMAND.get(command), args);
				processBuilder.redirectErrorStream(true); //合并输出流和错误流
				Process process = processBuilder.start();
				try(BufferedReader reader = new BufferedReader(new InputStreamReader(process.getInputStream()))){
					String line;
					while((line = reader.readLine())!=null) {
						resultBuilder.append(line+"\n");
					}
				}
				process.waitFor();
			} catch (IOException | InterruptedException e) {
				AlertUtil.exception(e);
			}
		}else {
			resultBuilder.append("命令【"+command+"】不被支持！\n\n");
			resultBuilder.append("以下是目前支持的命令列表：\n");
			for(String key:KAFKA_COMMAND.keySet()) {
				resultBuilder.append(key+"\n");
			}
			resultBuilder.append("\n");
		}
		log.info("执行结果："+resultBuilder.toString());
		return resultBuilder.toString();
	}

	@Override
	public void close() throws IOException {
		if(adminClient!=null) {
			adminClient.close();
		}
	}
}
